#include "redis/redis.hpp"
#include <iostream>

// 构造函数
Redis::Redis() : _publish_context(nullptr), _subscribe_context(nullptr)
{
}

// 析构函数
Redis::~Redis()
{
    if (_publish_context != nullptr)
    {
        redisFree(_publish_context);
    }
    if (_subscribe_context != nullptr)
    {
        redisFree(_subscribe_context);
    }
}

// 连接redis服务器
bool Redis::connect()
{
    // 1.得到redis上下文句柄
    _publish_context = redisConnect("127.0.0.1", 6379);
    if (_publish_context == nullptr || _publish_context->err)
    {
        if (_publish_context != nullptr)
        {
            std::cerr << "Error: " << _publish_context->err << std::endl;
            return false;
        }
        else
        {
            std::cerr << "Error: can't allocate redis context!" << std::endl;
            return false;
        }
    }
    // 1.得到redis上下文句柄
    _subscribe_context = redisConnect("127.0.0.1", 6379);
    if (_subscribe_context == nullptr || _subscribe_context->err)
    {
        if (_subscribe_context != nullptr)
        {
            std::cerr << "Error: " << _subscribe_context->err << std::endl;
            return false;
        }
        else
        {
            std::cerr << "Error: can't allocate redis context!" << std::endl;
            return false;
        }
    }
    // 2.在单独的线程中， 监听通道上的事件， 一有消息就给业务层进行上报
    std::thread t([&]()
                  { observer_channel_message(); });
    t.detach();

    std::cout << "Connected to redis-server success!" << std::endl;
    return true;
}

// 向redis指定的通道发布消息
bool Redis::publish(int channel, std::string message)
{
    // 1.防止在没有有效 Redis 连接的情况下尝试执行命令
    if (!_publish_context)
    {
        return false;
    }

    // 2.用于向 Redis 服务器发送命令： 将message消息 [发布] 到channelStr通道上
    // redisCommand <==> redisAppendCommand 写到本地缓冲中 + redisBufferWrite 从本地缓冲发送到redisserver + redisGetReply 阻塞等待服务应答
    std::string channelStr = std::to_string(channel);
    redisReply *reply = (redisReply *)redisCommand(_publish_context, "PUBLISH %s %s", channelStr.c_str(), message.c_str());
    if (reply == nullptr)
    {
        std::cout << "Failed to publish message: context error or disconnected.\n";
        return false;
    }
    // 3.根据回复判断是否成功
    bool success = (reply->type != REDIS_REPLY_ERROR);
    // 4.释放回复对象
    freeReplyObject(reply);

    return success;
}

// 向redis指定的通道订阅消息
bool Redis::subscribe(int channel)
{
    /* 这里只做订阅通道， 不接收通道消息
       通道消息的接收专门放到另外一个std::thread t()独立线程中进行
       这里只负责发送命令， 不会阻塞接收redis server响应的消息，
       否则会和notifyMsg抢占响应资源 */
    if (_subscribe_context == nullptr)
    {
        return false;
    }

    std::string channelStr = std::to_string(channel);
    if (REDIS_ERR == redisAppendCommand(_subscribe_context, "SUBSCRIBE %s", channelStr.c_str()))
    {
        std::cerr << "subscribe command failed!" << std::endl;
        return false;
    }

    // redisBufferWrite可以循环发送缓冲区， 直到缓冲区数据发送完毕(done被置为1)
    // 接收实际写入的字节数
    int written = 0;
    while(!written)
    {
        // 手动将当前连接的输出缓冲区中的命令数据写入到套接字, 在非阻塞模式下使用
        // 优化性能，特别是在需要处理大量命令或高并发请求的情况下非常有用。
        if (REDIS_OK != redisBufferWrite(_subscribe_context, &written))
        {
            std::cerr << "subscribe command failed!" << std::endl;
            return false;
        }
    }
    // REDIS_OK == redisBufferWrite()  完成从本地缓冲到服务器的订阅任务
    return true;
}

// 向redis指定的通道取消订阅消息
bool Redis::unsubscribe(int channel)
{
    // 1.向本地缓存发送取消订阅消息
    std::string channelStr = std::to_string(channel);
    if (REDIS_ERR == redisAppendCommand(_subscribe_context, "UNSUBSCRIBE %s", channelStr.c_str())) 
    {
        std::cerr << "unsubscribe command failed!" << std::endl;
        return false;
    }

    int written = 0;
    while (!written)
    {
        if (REDIS_ERR == redisBufferWrite(_subscribe_context, &written))
        {
            std::cerr << "unsubscribe command failed!" << std::endl;
            return false;
        }
    }
    return true;
}

// 在独立线程中接收订阅通道中的消息
void Redis::observer_channel_message()
{
    redisReply * reply = nullptr;
    // 子线程中 阻塞 接收订阅通道中的消息
    // (void **) &var    将一个变量的地址转换为通用指针类型 void*
    while (REDIS_OK == redisGetReply(_subscribe_context, (void **)&reply))
    {
        // 订阅收到的消息是一个带三元素的数组 reply->element[0]->str == "message", [1]->str == "通道" [2]->str == "消息内容"
        // reply!= nullptr 检查reply是否有效
        // reply->element[2]其第三个元素(即索引2)存在
        // reply->element[2]->str 包含非空字符串数据
        if (reply != nullptr && reply->element[2] != nullptr && reply->element[2]->str != nullptr)
        {
            // 通道上发生消息，那就通知上层业务
            _notify_message_handler(std::stoi(reply->element[1]->str), reply->element[2]->str);
        }

        // 每轮循环都释放redisReply结构
        // 每次调用 redisGetReply 都会分配一个新的 redisReply 结构，因此需要手动管理其生命周期
        freeReplyObject(reply);
    }
    std::cerr << ">>>>>>>>>>>>>>>>>>>>> observer_channel_message quit <<<<<<<<<<<<<<<<<<<<" << std::endl;
}

// 初始化向业务层上报通道消息的回调对象 设置回调的公有方法
void Redis::init_notify_handler(std::function<void(int, std::string)> fn)
{
    _notify_message_handler = fn;
}
